
                /*********************************************
                **                                          **
                **                                          **
                **              gp.d.nb.cml.C               **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/


/*
** Direct Access Global, Global Pointers and 
** Non-Blocking Global Read/Write Functions
** Intra Group communication
** Needs 4 processors
*/

#include "arch.h"

Global_id gid;
Global_id gid1;
Coll_id cid1;
Coll_id cid2;
Coll_id cid3;
Coll_id cid4;
Coll_id cid5;
Coll_id cid6;
Coll_id cid7;

void init_gid(int, char **){
   gid.set();
   gid1.set();
   cid1.set();
   cid2.set();
   cid3.set();
   cid4.set();
   cid5.set();
   cid6.set();
   cid7.set();
   fprintf(Sdf::trace, "from node %d: gid=%d, gid1=%d, cid1=%d, cid2=%d, cid3=%d, cid4=%d, cid5=%d, cid6=%d, cid7=%d\n", 
           Sdf::PROC_ID, gid.get(), gid1.get(), 
           cid1.get(), cid2.get(), cid3.get(),
           cid4.get(), cid5.get(), cid6.get(), cid6.get());
   fflush(Sdf::trace);
}


int comm_rank;
int comm_size;
CommL *comm;

int count;

void wcs_handler(HRW_rcd *ptr){
   count--;
   fprintf(Sdf::trace, "node%d: global-writing's over on call side\n", Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = ptr->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d: wcs_handler, rem_proc=%d, d_ptr=%d, g-sz=%d, th_ptr=%d, hndlr_ptr=%d, hndlr_d_ptr=%d, tag_dv=%d, atype=%d, tag_av=%d, tag_ev=%d, comm=%d, offset=%d, Tsize=%d\n",
           Sdf::PROC_ID, 
           ptr->get_remote_proc(),
           ptr->get_data_ptr(), 
           ptr->get_given_size(),
           ptr->get_thread_ptr(), 
           ptr->get_hndlr_ptr(),
           ptr->get_hndlr_data_ptr(),
           tag_dv,
           access_type,
           ptr->get_tag_av(),
           ptr->get_tag_ev(),
           ptr->get_comm(),
           ptr->get_offset(),
           ptr->get_Tsize());
   fflush(Sdf::trace);
}

void rcs_handler(HRW_rcd *ptr){
   count--;
   fprintf(Sdf::trace, "node%d: global-reading's over on call side\n",Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = ptr->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d: rcs_handler, rem_proc=%d, d_ptr=%d, g-sz=%d, r_sz=%d, th_ptr=%d, hndlr_ptr=%d, hndlr_d_ptr=%d, tag_dv=%d, atype=%d, tag_av=%d, tag_ev=%d, comm=%d, offset=%d, Tsize=%d\n",
           Sdf::PROC_ID, 
           ptr->get_remote_proc(),
           ptr->get_data_ptr(), 
           ptr->get_given_size(), 
           ptr->get_rcvd_size(), 
           ptr->get_thread_ptr(), 
           ptr->get_hndlr_ptr(), 
           ptr->get_hndlr_data_ptr(),
           tag_dv,
           access_type,
           ptr->get_tag_av(),
           ptr->get_tag_ev(),
           ptr->get_comm(),
           ptr->get_offset(),
           ptr->get_Tsize());
   fflush(Sdf::trace);
}

class Root_thread: public Thread{
   public:
        Root_thread()
                : Thread((Tmf0_ptr)&Root_thread::body){}

        void body(){

           fprintf(Sdf::trace, "from node%d: this is my rank\n", Sdf::PROC_ID);
           fflush(Sdf::trace);

           int colour;
           if(Sdf::PROC_ID==Sdf::PROC_NB-1)
              colour=Sdf::UNDEFINED;
           else
              colour=1;

             //Sdf::UNDEFINED is not counted as a colour
           CommL **comm_table = 
           Sdf::COMML_WORLD->split(cid1, 1, colour, Sdf::PROC_NB-2-Sdf::PROC_ID);
           comm = *comm_table;
 
           if(Sdf::PROC_ID==Sdf::PROC_NB-1){

              Sdf::barrier(cid2);
                //synchronizes all processes in Sdf::COMML_WORLD
              fprintf(Sdf::trace, "node%d: !!!!!!!!!!1\n", Sdf::PROC_ID);
              fflush(Sdf::trace);
              char data[21];

              Global<char> global1(gid1, (char *)0, Sdf::COMML_WORLD, comm);
              Sdf::barrier(cid3);

              Star<char> global1_ptr(Sdf::NOPID, global1,
                                     (unsigned long)0, Sdf::NORA, Sdf::NOAP, Sdf::NOCP);

                //last process in COMML_WORLD reads the result of global
                //read/write performed by the other over comm. j stands
                //for the process id in comm.
              for(int j=0; j<Sdf::PROC_NB-1; j++){
                 global1_ptr.set_proc_id(j);
                 (*global1_ptr).read(data[0], 21);
                 for(int i=0; i<21; i++)
                    fprintf(Sdf::trace, "node%d:  data[%d]=%d, read from: rk-in-world=%d, rk-in-comm=%d\n", 
                            Sdf::PROC_ID, i, data[i], Sdf::PROC_NB-2-j,j);
                 fflush(Sdf::trace);
              }

              Sdf::barrier(cid4);
                //release all processes in comm that were running
                //while the last one was reading

              delete comm;
              return;
           }

           comm_rank = comm->rank();
           comm_size = comm->size();

           fprintf(Sdf::trace, "node%d, in Root_thread::body2 rank=%d, comm_size=%d\n",
                   Sdf::PROC_ID, comm_rank, comm_size);
           fflush(Sdf::trace);

           char source_data[21];
           char rw_target[21];
           char dest_data[21];

           int s;
           for(s=0; s<21; s++)
              source_data[s]  = comm_rank;
           for(s=0; s<21; s++)
              rw_target[s]    = 50+comm_rank;
           for(s=0; s<21; s++)
              dest_data[s]    = 100+comm_rank;

           int i;
           for(i=0; i<21; i++)
             fprintf(Sdf::trace, "node%d: s[%d]=%d, t[%d]=%d, d[%d]=%d\n",
                     Sdf::PROC_ID, i, source_data[i], i, 
                     rw_target[i], i, dest_data[i]);
           fflush(Sdf::trace);

           Global<char> global(gid, rw_target, comm);
           Sdf::barrier(cid5, comm);
             //synchronizes global construction over comm

           {Sdf::Access_type access_type;
            int tag_dv = global.get_tag_dv(&access_type);
            fprintf(Sdf::trace, "node%d, test global accessors, tag_ev=%d, tag_dv=%d, access_type=%d, tag_av=%d, direct_handle=%d, comm=%d\n",
                    Sdf::PROC_ID,
                    global.get_tag_ev(),
                    tag_dv,
                    access_type,
                    global.get_tag_av(),
                    global.get_direct_handle(),
                    global.get_comm());
            fflush(Sdf::trace);
           }

           count = 0;

           RWCtrl ctrl;
           Star<char> global_ptr((comm_rank+1)%comm_size, global, 
                                 (unsigned long)0, Sdf::NORA, Sdf::NOAP, &ctrl);

             //writing phase via global
           ctrl.set_hndlr_ptr(wcs_handler);
           ctrl.set_hndlr_data_ptr(0);
           (*global_ptr).write(source_data[0], 21); 

           count++;
           while(count)
              Sdf::poll();
                  //achieves local completion
           Sdf::barrier(cid6, comm);
                  //achieves global completion

             //reading phase via global
           ctrl.set_hndlr_ptr(rcs_handler);
           ctrl.set_hndlr_data_ptr(0);
           global_ptr.set_proc_id((comm_rank+comm_size-1)%comm_size);
           (*global_ptr).read(dest_data[0], 21);

           count++;
           while(count)
              Sdf::poll();
                  //achieves local completion
           Sdf::barrier(cid7, comm);
                  //achieves global completion

             //write-read result reporting
           for(i=0; i<21; i++)
             fprintf(Sdf::trace, "node%d: s[%d]=%d, t[%d]=%d, d[%d]=%d\n",
                     Sdf::PROC_ID, i, source_data[i], i, 
                     rw_target[i], i, dest_data[i]);
           fflush(Sdf::trace);

           Sdf::barrier(cid2);
             //synchronizes all processes in Sdf::COMML_WORLD
           fprintf(Sdf::trace, "node%d: !!!!!!!!!!2\n", Sdf::PROC_ID);
           fflush(Sdf::trace);

           Global<char> global1(gid1, dest_data, Sdf::COMML_WORLD, comm);
           Sdf::barrier(cid3);
             //synchronizes global1 construction over Sdf::COMML_WORLD

           Sdf::barrier(cid4);
             //keep all processes in comm running while the last
             //process performs reading

           delete comm;
        }
};

MAIN(Root_thread, init_gid, Sdf::empty, Sdf::empty)
